package com.shujia.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

object Demo4Trans {
  def main(args: Array[String]): Unit = {


    /**
      * 1、创建生产者
      *
      */

    val properties = new Properties()

    //1、kafka broker列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")

    //2、指定kv的序列化类
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    //指定事务id
    properties.setProperty("transactional.id", "123123123")


    val kafkaProducer = new KafkaProducer[String, String](properties)


    /**
      * 初始化事务
      *
      */

    //初始化事务
    kafkaProducer.initTransactions()
    //开始事务
    kafkaProducer.beginTransaction()

    //生产数据
    //虽然事务还没有提交，但是数据以及发送到kafka中了
    kafkaProducer.send(new ProducerRecord[String, String]("trans", "java"))

    Thread.sleep(10000)

    kafkaProducer.send(new ProducerRecord[String, String]("trans", "spark"))

    //提交事务,只是事务提交之后数据才会再kafka集群中生效
    kafkaProducer.commitTransaction()


    kafkaProducer.close()

    /**
      *
      * --isolation-level read_committed ： 读已提交数据
      *
      * kafka-console-consumer.sh --bootstrap-server master:9092,node1:9092,node2
      * ginning --isolation-level read_committed --topic trans
      *
      */

  }

}
